package com.bj58.ecdata.hadoop.fourthinfo;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.Serializable;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import com.bj58.data.source.infodetail.InfoDetail;

public class User_Post {
//	private static Pattern p = Pattern.compile("^cate(first|second)(\\d+)_local(first|second|third)(\\d+)_(.*?)(\\d{8})_([iu]count:\\d+)");
	private static final String input="/dsap/rawdata/infodetail/yyyyMMdd/";
	private static final String output="/dsap/middata/user_post_job/yyyyMMdd/";
	private static String sep = "|";
	public static String date = "";
	public static void setDate(String date1){
		date = date1;
	}
	public static class UserPostMapper extends Mapper<LongWritable, Text, Text, Text> {
		private Text mapOutputKey = new Text();
		private Text mapOutputValue = new Text();
		
		public void map(LongWritable key, Text value, Context context) {
			try {
				InfoDetail info = new InfoDetail(value.toString());
				
				String info_id = info.getInfoID();
				String user_id = info.getUserID();
				
				String local_path = info.getLocalPath();
				String[] locals = local_path.split(",");
				String local_id=locals[locals.length-1];
				
				String cate_path = info.getCatepath();
				String[] cates = cate_path.split(",");
				String cate_id = cates[cates.length-1];
				
				String add_time = String.valueOf(info.getAddTimeStamp());
				String operation = info.getUpType();
				String state = info.getState();
				String ip = info.getUserip();
				String phone = info.getPhone();
				String[] fields = {info_id,user_id,local_id,cate_id,add_time,operation,state,ip,phone," "};
				String resultLine = StringUtils.join( fields,"@#");
				mapOutputKey.set(user_id);
				mapOutputValue.set(resultLine);
				context.write(mapOutputKey, mapOutputValue);
			} catch (Exception e) {
				e.printStackTrace();
				return;
			}
		}
	}
	
	public static class UserPostReducer extends Reducer<Text, Text, NullWritable, Text> {
		private MultipleOutputs<NullWritable, Text> mos;
		private HashMap<String,String> localInfos=null;
		private HashMap<String,CateInfo> cateInfos=null;
		private NullWritable nullKey=NullWritable.get();
		private Text outputValue=new Text();
		private String date = "";
		public void setup(Context context) {
			try {
				date = context.getConfiguration().get("DATE");
				mos = new MultipleOutputs<NullWritable, Text>(context);
				BufferedReader br = new BufferedReader(new InputStreamReader(new FileInputStream("BelongCate.part"),"UTF-8"));
				cateInfos=getCateInfo(br);
				br = new BufferedReader(new InputStreamReader(new FileInputStream("BelongLocal.part"),"UTF-8"));
				localInfos=getLocalInfo(br);
				System.out.println(localInfos.size()+" localInfo!");
				System.out.println(cateInfos.size()+" cateInfo!");
			} catch (IOException e) {
				e.printStackTrace();
				return;
			}
		}
		@Override
		public void cleanup(Context context) {
			try {
				mos.close();
			} catch (Exception e) {
				e.printStackTrace();
				return;
			}
		}
		public static HashMap<String,String> getLocalInfo(BufferedReader reader) throws IOException{
			String line=null;
			HashMap<String,String> locals=new HashMap<String,String>();
			while((line=reader.readLine())!=null){
				String fields[]=line.split("\t");
				String localid=fields[0];
				String localname=fields[2];
				locals.put(localid, localname);
			}
			reader.close();
			return locals;
		}
		
		public static HashMap<String,CateInfo> getCateInfo(BufferedReader reader) throws IOException{
			String line=null;
			HashMap<String,CateInfo> cateInfos=new HashMap<String,CateInfo>();
			while((line=reader.readLine())!=null){
				String fields[]=line.split("\t");
				String cateid=fields[0];
				String catename=fields[2];
				CateInfo cate=new CateInfo(catename);
				String levels[]=fields[5].split(",");
				int len=levels.length<2?levels.length:2;
				for(int i=0;i<len;i++){
					cate.setCateids(i, levels[i]);
				}
				cateInfos.put(cateid, cate);
			}
			reader.close();
			return cateInfos;
		}
		
		public String getCateStr(String cateID){
			CateInfo  cate=cateInfos.get(cateID);
			if(cate == null)
				return sep;
			String[] catesIds=cate.getCateids();
			String cateStr = "";
			for(int i=0;i<2;i++){
				if(catesIds[i].equals("0")||!cateInfos.containsKey(catesIds[i]))
					cateStr += "";
				else
					cateStr += cateInfos.get(catesIds[i]).getCatename();
				if(i == 0)
					cateStr += sep;
			}
			return cateStr;
		}
		public void reduce(Text key, Iterable<Text> values, Context context) {
			HashMap<String,Integer> city_count = new HashMap<String,Integer>();
			HashMap<String,String> city_time = new HashMap<String,String>();
			HashMap<String,Integer> cate_count = new HashMap<String,Integer>();
			HashMap<String,String> cate_time = new HashMap<String,String>();
			HashMap<String,Integer> phone_count = new HashMap<String,Integer>();
			HashMap<String,String> phone_time = new HashMap<String,String>();
			int post_count = 0;
			int modify_post_count = 0;
			int delete_post_count = 0;
			//处理一个user_id的行集
			for(Text t:values){
				String[] fields = t.toString().split("@#");
				int operation = 0;
				int state = 0;
				try{
					operation = Integer.parseInt(fields[5]);
					state = Integer.parseInt(fields[6]);
				}catch(Exception e) {
				}
				if(operation==0){
					String city = localInfos.containsKey(fields[2]) ? localInfos.get(fields[2]) : "";
					int city_post_count = city_count.containsKey(city) ? city_count.get(city): 0;
					city_count.put(city, city_post_count+1);
					city_time.put(city, fields[4]);
					String cate = getCateStr(fields[3]);
					int cate_post_count = cate_count.containsKey(cate) ? cate_count.get(cate) : 0;
					cate_count.put(cate, cate_post_count+1);
					cate_time.put(cate, fields[4]);
					
					String[] phones=fields[8].split("[^\\d-]");
					for(String phone : phones){
						if(phone.trim() == "")
							continue;
						int phone_post_count = phone_count.containsKey(phone) ? phone_count.get(phone) : 0;
						phone_count.put(phone, phone_post_count+1);
						phone_time.put(phone, fields[4]);
					}
					post_count += 1;
				}else if(operation==1)
					modify_post_count += 1;
				if(state == 0){
					delete_post_count += 1;
				}
			}
			
			// output_post
			if(post_count > 0 || modify_post_count > 0 || delete_post_count > 0){
				String city = getMax(city_count);
				String cates = getMax(cate_count);
				if(cates == "")
					cates = sep;
				String phone = getMax(phone_count);
				
				String outFields[] = {date, key.toString(), city, cates, phone, String.valueOf(post_count), String.valueOf(modify_post_count), String.valueOf(delete_post_count)};
				outputValue.set(StringUtils.join(outFields,sep));
				try {
					mos.write("multiple",nullKey, outputValue, "user_post/part");
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
			
			//output_attrs
			if(post_count > 0){
				String city = get_attrs(city_count,city_time);
				String cates = get_attrs(cate_count,cate_time);
				String phone = get_attrs(phone_count,phone_time);
				String outFields[] = {date, key.toString(), city, cates, phone};
				outputValue.set(StringUtils.join(outFields, sep));
				try {
					mos.write("multiple",nullKey, outputValue, "user_attrs/part");
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}
		
		private String  getMax(HashMap<String,Integer> countMap){
			Iterator<String> iter = countMap.keySet().iterator();
			String finalKey = "";
			int maxVal = 0;
			while(iter.hasNext()){
				String key = iter.next();
				int count = countMap.get(key);
				if(count >= maxVal)
					finalKey = key;
			}
			return finalKey;
		}
	
		private String  get_attrs(HashMap<String,Integer> countMap,HashMap<String,String> timeMap){
			Iterator<String> iter = countMap.keySet().iterator();
			List<String> rs = new ArrayList<String>();
			while(iter.hasNext()){
				String key = iter.next();
			    String time = timeMap.get(key);
			    int count = countMap.get(key);
				if(key.indexOf(sep) >= 0){
					key = key.replace(sep, "_");
				}
				rs.add(key+":"+count+","+time);
			}
			return StringUtils.join(rs,"#");
		}
	}
	
	
	public static class CateInfo implements Serializable{
		/**
		 * 
		 */
		private static final long serialVersionUID = 2L;
		private String catename;
		private String[] cateids=new String[2];
		
		public CateInfo(String catename) {
			this.catename=catename;
			for(int i=0;i<2;i++)
				cateids[i]="0";
		}
		public void setCateIds(String id1,String id2){
			cateids[0]=id1;
			cateids[1]=id2;
		}
		public String getCatename() {
			return catename;
		}
		public void setCatename(String catename) {
			this.catename = catename;
		}
		public String[] getCateids() {
			return cateids;
		}
		public void setCateids(String[] cateids) {
			this.cateids = cateids;
		}
		public void setCateids(int index,String id) {
			cateids[index] = id;
		}
	}
	
	  public static boolean checkPath(String dirPath) throws IOException
	   {
	       Configuration conf = new Configuration();
	       FileSystem fs = FileSystem.get(conf);
	       Path path=new Path(dirPath);
	       boolean exists  = fs.exists(path);
	       
	       fs.close();
	       return exists;
	   }
		 /**删HDFS文件夹
	    * @param dirPath  路径
	    * @param isNull	true:非空文件夹也删除   false:只删除空文件夹
	    * @throws IOException
	    */
	   public static boolean deleteDir(String dirPath,boolean isNull) throws IOException
	   {
	       Configuration conf = new Configuration();
	       FileSystem fs = FileSystem.get(conf);
	       Path path=new Path(dirPath);
	       boolean isDeleted  = fs.delete(path, isNull);
	       
	       fs.close();
	       return isDeleted;
	   }
	   
	public static int hadoopMainEntrance(String statDate) throws Exception{
		statDate=DateUtil.strToStr(statDate, "yyyyMMdd");
		User_Post.setDate(statDate);
		System.out.println(date);
		Configuration conf = new Configuration();		
		Path catePath = new Path("/dsap/rawdata/cmc_category/" + statDate
				+ "/part-m-00000");
		Path localPath = new Path("/dsap/rawdata/cmc_local/" + statDate
				+ "/part-m-00000");
		String outPath1UriWithLink = catePath.toUri().toString() + "#"
				+ "BelongCate.part";// #号之后的名称是对上面文件的链接，不同文件的链接名必须不能相同
		String outPath2UriWithLink = localPath.toUri().toString() + "#"
				+ "BelongLocal.part";
		try {
			DistributedCache.addCacheFile(new URI(outPath1UriWithLink), conf);
		} catch (URISyntaxException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		DistributedCache.addCacheFile(new URI(outPath2UriWithLink), conf);

		int returnCode = 126;
		conf.set("DATE", statDate);
		Job job = new Job(conf, "user_post_job");
		job.setJarByClass(User_Post.class);
		job.setMapperClass(UserPostMapper.class);
		job.setReducerClass(UserPostReducer.class);
		job.setNumReduceTasks(20);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);
		job.setOutputKeyClass(NullWritable.class);
		job.setOutputValueClass(Text.class);
		MultipleOutputs.addNamedOutput(job, "multiple", TextOutputFormat.class,
				 NullWritable.class, Text.class);
		if (FileSystem.get(conf).exists(new Path(input.replaceAll("yyyyMMdd", statDate)))) {
			FileInputFormat.addInputPath(job, new Path(input.replaceAll("yyyyMMdd", statDate)));
		} else {
			return -1;
		}
		String outPath=output.replaceAll("yyyyMMdd", statDate);
		if(checkPath(outPath)){
			deleteDir(outPath, true);
		}
		FileOutputFormat.setOutputPath(job, new Path(outPath));
		returnCode = job.waitForCompletion(true) ? 0 : 1;
		return returnCode;
	}
	
	/**
	 * @param args
	 */
	public static void main(String[] args) {
//		String[] fields = {"123","456"," ",null,"789"};
//		System.out.println(StringUtils.join(fields,"@#"));
		String startDate = "";
		String endDate = "";
		
		if(args.length > 0 ){
			if(args.length == 1 ){
				startDate = args[0];
				endDate = args[0];
			}else if(args.length == 2 ){
				startDate = args[0];
				endDate = args[1];
			}
		}
		
		if(startDate == null || "".equals(startDate)){
			startDate = DateUtil.getStringDate(DateUtil.getDateBefore(new Date(), 1), "yyyy-MM-dd");
			endDate = startDate;
		}
		
		for(String statDate:DateUtil.getDateList(startDate, endDate)){
			System.out.println("deal:" + statDate);
            try {
				hadoopMainEntrance(statDate);
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}

}
